package com.miao.im.repository;

import cn.hutool.core.collection.CollectionUtil;
import com.alibaba.fastjson.JSON;
import com.miao.im.common.CacheKey;
import com.miao.im.common.enums.LuaName;
import com.miao.im.service.message.ChatMessage;
import com.miao.im.service.message.OfflineMessageCounter;
import com.miao.im.support.RedisLuaExecutor;
import com.miao.im.support.util.JsonUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.ReactiveStringRedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Repository;
import reactor.core.publisher.Mono;

import java.time.ZoneOffset;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @Auther: miaoguoxin
 * @Date: 2021/5/19 20:39
 * @Description:
 */
@Repository
@Slf4j
public class MessageRepository implements IMessageRepository {
    @Autowired
    private StringRedisTemplate redisTemplate;
    @Autowired
    private RedisLuaExecutor redisLuaExecutor;
    @Autowired
    private ReactiveStringRedisTemplate reactiveStringRedisTemplate;

    @Override
    public Mono<Void> reactiveAddOfflineMsg(ChatMessage chatMessage) {
        String json = JsonUtil.toJSONStr(chatMessage);
        String key = getKey(chatMessage.getFromUserId(), chatMessage.getTargetId(), chatMessage.getTargetType());
        List<String> keys = CollectionUtil.newArrayList(key);
        long score = chatMessage.getSendTime().toInstant(ZoneOffset.of("+8")).toEpochMilli();
        Object[] args = new Object[]{
                String.valueOf(CacheKey.OFFLINE_MSG_MAX_SIZE - 1),
                String.valueOf(score),
                json,
                String.valueOf(System.currentTimeMillis())
        };
        return redisLuaExecutor.reactiveExecute(LuaName.SAVE_OFFLINE_MSG, keys, args).then();
    }

    @Override
    public Mono<Void> reactiveIncOfflineCounter(ChatMessage chatMessage) {
        String key = CacheKey.OFFLINE_MSG_COUNTER_PREFIX + chatMessage.getTargetId();
        String hashKey = chatMessage.getTargetType() + CacheKey.SEMICOLON_DELIMITER + chatMessage.getFromUserId();
        return reactiveStringRedisTemplate.opsForHash().increment(key, hashKey, 1)
                .then(this.saveMsgCounter(key, hashKey, chatMessage));
    }

    @Override
    public Map<String, OfflineMessageCounter> getMsgCounter(Long userId) {
        String key = CacheKey.OFFLINE_MSG_COUNTER_PREFIX + userId;
        Map<Object, Object> map = redisTemplate.opsForHash().entries(key);
        Map<String, Integer> counterMap = new HashMap<>();
        Map<String, OfflineMessageCounter> result = new HashMap<>();
        for (Map.Entry<Object, Object> entry : map.entrySet()) {
            String oKey = entry.getKey().toString();
            String val = entry.getValue().toString();
            if (oKey.endsWith("msg")) {
                OfflineMessageCounter counter = JSON.parseObject(val, OfflineMessageCounter.class);
                result.put(oKey, counter);
            } else {
                counterMap.put(oKey, Integer.parseInt(val));
            }
        }
        for (Map.Entry<String, Integer> entry : counterMap.entrySet()) {
            OfflineMessageCounter messageCounter = result.get(entry.getKey() + CacheKey.SEMICOLON_DELIMITER + "msg");
            if (messageCounter != null) {
                messageCounter.setCount(entry.getValue());
            }
        }
        return result;
    }

    private static String getKey(Long userId, Long targetId, Integer targetType) {
        return CacheKey.OFFLINE_MSG_PREFIX + targetId +
                CacheKey.SEMICOLON_DELIMITER + targetType +
                CacheKey.SEMICOLON_DELIMITER + userId;
    }

    private Mono<Void> saveMsgCounter(String key, String hashKey, ChatMessage chatMessage) {
        OfflineMessageCounter counter = new OfflineMessageCounter(chatMessage);
        String msgKey = hashKey + CacheKey.SEMICOLON_DELIMITER + "msg";
        return reactiveStringRedisTemplate.opsForHash().put(key, msgKey, JsonUtil.toJSONStr(counter)).then();
    }
}
